pyrobot/0000775000076500007650000000000010273263467012052 5ustar dblankdblankpyrobot/compile.py0000775000076500007650000000020410271002540014030 0ustar dblankdblankimport compileall __author__ = "Douglas Blank " __version__ = "$Revision: 1.2 $" compileall.compile_dir(".") pyrobot/brain/0000775000076500007650000000000010273263463013141 5ustar dblankdblankpyrobot/brain/rbf.py0000664000076500007650000000271210271002541014250 0ustar dblankdblank # A Radial Basis Function (unit) has three things that can be learned: # center - the model vector # radius - radius of the function # weight - weighting of this node __author__ = "Douglas Blank " __version__ = "$Revision: 1.2 $" import Numeric, math, random # Typical activation functions are: def gaussian(v): return map(math.exp, -(v ** 2)) def imqe(v, k = .5): return map(lambda z: 1.0 / math.sqrt(z + (k ** 2)), v) class RBF: """ A Radial Basis Function node """ def __init__(self, size): # center can be selected at random from training data. Here # we just assign a random vector: self.center = Numeric.array([random.random() for r in range(size)]) # initial radius: self.radius = 1.0 # initial values of activation: self.activation = Numeric.zeros(size) def propagate(self, input): self.activation = gaussian((input - self.center) / self.radius) class RBFLayer: """ A layer of RBF units """ def __init__(self, size, vsize): self.units = [RBF(vsize) for i in range(size)] def propagate(self, input): for u in self.units: u.propagate(input) # make a layer of 10 units, each capable of storing a 5-length # model vector: rbfLayer = RBFLayer(10, 5) # sample vector: input = Numeric.array([0.0, 0.2, 0.4, 0.3, 0.8]) # compare it to units: rbfLayer.propagate(input) for u in rbfLayer.units: print u.activation pyrobot/brain/td.py0000664000076500007650000001006410272077124014117 0ustar dblankdblank""" ---------------------------------------------------- A temporal differencing backprop. Part of the Pyrobot Robotics Project. Provided under the GNU General Public License. ---------------------------------------------------- (c) 2001-2005, Developmental Robotics Research Group ---------------------------------------------------- This file implements the major classes and functions for performing Temporal Differencing in Python. Part of the Pyrobot project. """ __author__ = "Robert Casey " __version__ = "$Revision: 1.1 $" from Numeric import resize import itertools class Temporal_Difference: """ Class which contains Utility Values and performs operations on them """ # constructor def __init__(self, sizex, sizey, goal, pits, alpha=0.2): self.goal = goal self.pits = pits self.num_squares_x = sizex self.num_squares_y = sizey self.alpha = alpha # self.utility = resize( 0.0,(self.num_squares_x,self.num_squares_y)); self.reset_utils() # self.reward = resize( 0.0,(self.num_squares_x,self.num_squares_y)); self.reset_reward() # self.frequencies = resize( 0,(self.num_squares_x,self.num_squares_y)); self.reset_frequencies() def reset_utils(self): self.utility = resize(0.0,(self.num_squares_x,self.num_squares_y)); for e in self.pits: self.utility[e] = -1.0 e = self.goal self.utility[e] = 1.0 def reset_reward(self): self.reward = resize(0.0,(self.num_squares_x,self.num_squares_y)) self.reward[self.goal] = 100 for e in self.pits: self.reward[e] = -50 def reset_frequencies(self): self.frequencies = resize(0,(self.num_squares_x,self.num_squares_y)) def get_alpha( self ): return self.alpha def set_alpha( self, new_val ): self.alpha = new_val def running_average( self, util, reward, freq ): return ((util * (freq - 1) + reward) / freq) def do_td( self, p ): U = self.utility frequencies = self.frequencies final_states = self.pits + [self.goal] # we want to traverse in reverse! p.reverse(); next_state = p[0] # start from end, go until start for curr_state in p: # update the frequencies of all states in the path frequencies[curr_state] = frequencies[curr_state] + 1 if( curr_state in final_states ): rew = self.reward[curr_state] - (len(p) * 0) # 2) U[curr_state] = round( self.running_average( U[curr_state], self.reward[curr_state], frequencies[curr_state] ), 10) ; else: temp = U[next_state]-U[curr_state]; U[curr_state] = round(U[curr_state] + self.alpha*(self.reward[curr_state] + temp), 10); # since we're iterating backwards next_state = curr_state; self.frequencies = frequencies self.utility = U def get_utility( self, x, y ): return self.utility[x][y] # returns an RGB string for a given utility value def get_utility_color(self, x, y): util = self.utility[x][y] if util < 0: neg = 1 util *= -1 else: neg = 0 r = 255 g = 255 - int( util * 2.55 ) b = 255 - ( int( util * 2.55 ) * 2 ) if b < 0: b = 0 # print "%d:%d:%d" % ( r,g,b ) if neg: tmp = b b = r r = tmp strr = hex(r)[2:] while len(strr) < 2: strr = "0" + strr; strg = hex(g)[2:] while len(strg) < 2: strg = "0" + strg; strb = hex(b)[2:] while len(strb) < 2: strb = "0" + strb; strcolor = "#" + strr + strg + strb return( "c_%2d%2d_%s" % (x, y, strcolor) ) pyrobot/brain/behaviors/0000775000076500007650000000000010273263463015123 5ustar dblankdblankpyrobot/brain/behaviors/__init__.py0000775000076500007650000002754710262150774017254 0ustar dblankdblank# This is basically a rewrite of XRCL in Python # BehaviorBased Brain # defines the Behavior-based brain, behaviors, and states import time from pyrobot.brain import Brain class BehaviorBasedBrain(Brain): """ This is the main engine that runs collections of behaviors (states). Usually, you create once of these per robot. """ def __init__(self, controllers = {}, engine = 0): Brain.__init__(self, 'BehaviorBasedBrain', engine) self.states = {} self.controls = controllers self.history = [{}, {}, {}] self.pie = [] self.desires = [] self.effectsTotal = {} self.initialized = 0 self.activeState = None def set_controls(self, controllers): self.controls = controllers self.history = [{}, {}, {}] def activate(self, name): self.states[name].status = 1 self.states[name].onActivate() self.goto(name) def goto(self, name): self.activeState = self.states[name] def deactivate(self, name): self.states[name].status = 0 self.states[name].onDeactivate() self.activeState = None def add(self, state): if state.name in self.states.keys(): raise "ERROR: state already exists: '" + state.name + "'" self.states[state.name] = state state.engine = self.engine state.robot = self.engine.robot state.brain = self state.setup() if state.status: state.onActivate() def reset(self): self.states = {} #self.status = -1 self.desires = [] self.effectsTotal = {} def step(self): if not self.initialized: for s in self.states.keys(): self.states[s].setcontrols(self.controls) self.initialized = 1 self.desires = [] # init all desires (this will be set in state.run) self.history[2] = self.history[1] self.history[1] = self.history[0] self.history[0] = {} self.effectsTotal = {} for s in self.states.keys(): if self.states[s].status == 1: self.states[s].run() # desires: truth, controller, value, rulename, behname, statename control = {} # set all totalTruths to 0, totalEffects to 0 totalTruth = {} totalEffects = {} for c in self.controls.keys(): totalTruth[c] = 0.0 totalEffects[c] = 0.0 for e in self.effectsTotal.keys(): # state:beh:controller s, b, c = e.split(':') totalEffects[c] = max(float(self.effectsTotal[e]), totalEffects[c]) # sum up totalTruth for d in self.desires: # compute total truth for each controller totalTruth[d[1]] += d[0] * (self.effectsTotal[d[5]+":"+d[4]+":"+d[1]] / totalEffects[d[1]]) self.pie = [] for d in self.desires: # (beffect / totaleffect) * (truth / totaltruth) * value c = d[1] if totalTruth[c] != 0: part = ((d[0]*(self.effectsTotal[d[5]+":"+d[4]+":"+d[1]]/totalEffects[d[1]]))/totalTruth[c]) else: part = 0 amt = part * d[2] self.pie.append( [d[1], (self.effectsTotal[d[5] + ":" + d[4] + ":" + c] / totalEffects[c]), part, d[2], amt, d[5] + ":" + d[4] + ":" + d[3] ] ) if c in control.keys(): control[c] += amt else: control[c] = amt for c in self.controls.keys(): if c in control.keys(): # set that controller to act with a value #print "setting %s to value %f" % (c, control[c]) self.controls[c](control[c]) self.history[0][c] = control[c] # ------------------------------------------------- # This will update robot's position so that the GUI # can draw it, even if no command is sent to move # the robot. # ------------------------------------------------- self.controls['update']() # ------------------------------------------------- # let's force a 0.1s break so that we are going to have # at most 10 cycles per second # an improved version is expected # ------------------------------------------------- time.sleep(0.01) # change states' status if necessary for s in self.states.keys(): for d in self.states[s].deactivatelist: #deactivate first self.deactivate(d) for a in self.states[s].activatelist: self.activate(a) self.states[s].activatelist = [] self.states[s].deactivatelist = [] def stop_all(self): for c in self.controls.keys(): # set that controller to act with a value self.controls[c](0) def redrawPie(self, pie, percentSoFar, piececnt, controller, percent, name): # FIX: behavior specific. How to put in Behavior-based code? xoffset = 5 yoffset = 20 width = 100 row = (pie - 1) * (width * 1.5) colors = ['blue', 'red', 'tan', 'yellow', 'orange', 'black', 'azure', 'beige', 'brown', 'coral', 'gold', 'ivory', 'moccasin', 'navy', 'salmon', 'tan', 'ivory'] self.canvas.create_text(xoffset + 60,row + 10, tags='pie',fill='black', text = controller) self.canvas.create_arc(xoffset + 10,row + yoffset,width + xoffset + 10,row + width + yoffset,start = percentSoFar * 360.0, extent = percent * 360.0 - .001, tags='pie',fill=colors[(piececnt - 1) % 17]) self.canvas.create_text(xoffset + 300,row + 10 + piececnt * 20, tags='pie',fill=colors[(piececnt - 1) % 17], text = name) def redraw(self): if len(self.pie) != 0: self.canvas.delete('pie') piecnt = 0 for control in self.controls: if control == "update": continue piecnt += 1 percentSoFar = 0 piececnt = 0 for d in self.pie: if control == d[0]: piececnt += 1 portion = d[2] #try: self.redrawPie(piecnt, percentSoFar, \ piececnt, \ "%s effects: %.2f" % \ (d[0], self.history[0][d[0]]), portion, \ "(%.2f) %s IF %.2f THEN %.2f = %.2f" % \ (d[1], d[5], d[2], d[3], d[4])) #except: # pass percentSoFar += portion else: self.canvas.create_text(200,130, tags='pie',fill='black', text = "Ready...") class Behavior: """ The core object. This gets subclassed for each beh instance """ def __init__(self, status = 0, effects = {}, name = ''): self.status = status self.type = self.__class__.__name__ self.name = name or self.type self.effects = effects self.robot = 0 # filled in later self.brain = 0 # filled in later self.engine = 0 # filled in later def setup(self): pass # this will get over written, normally def onActivate(self): pass def onDeactivate(self): pass def update(self): pass # this will get over written, normally def Effects(self, controller, amount = 1.0): self.effects[controller] = amount def IF(self, fvalue, controller, amount = 1.0, name = ''): if name == '': name = "Rule%d" % (len(self.rules) + 1) self.rules.append([float(fvalue), controller, float(amount), name]) class FSMBrain(BehaviorBasedBrain): """ This is the main engine that runs the FSM. """ def activate(self, name): self.states[name].status = 1 self.states[name].onActivate() def deactivate(self, name): self.states[name].status = 0 self.states[name].onDeactivate() def reset(self): self.states = {} def step(self): for s in self.states.keys(): if self.states[s].status == 1: self.states[s].run() # pause? for s in self.states.keys(): for d in self.states[s].deactivatelist: #deactivate first self.deactivate(d) for a in self.states[s].activatelist: self.activate(a) self.states[s].activatelist = [] self.states[s].deactivatelist = [] class State: """ Collections of behaviors. this gets subclassed by each collection """ def __init__(self, status = 0, name = ''): self.debug = 0 self.behaviors = {} self.activatelist = [] self.deactivatelist = [] self.status = status self.type = self.__class__.__name__ self.name = name or self.type def getState(self, statename): if statename in self.brain.states.keys(): return self.brain.states[statename] else: raise "ERROR: no such statename" def goto(self, state, *args): if self.debug: print "Leaving state '%s'; going to state '%s'..." % (self.name, state) self.deactivate(self.name) self.activate(state) self.brain.states[state].onGoto(args) self.brain.goto(state) def onGoto(self, args = []): # FIX: could make a nice way of setting class vars here. # Currently: # if you pass this vars, you must take care of them! pass # normally will overload def activate(self, name): if not (name in self.activatelist): self.activatelist.append(name) def deactivate(self, name): if not (name in self.deactivatelist): self.deactivatelist.append(name) def onActivate(self): pass def setup(self): pass # normally will overload def onDeactivate(self): pass # normally will overload def update(self): self.step() def step(self): pass # normally will overload def add(self, b): if b.name in self.behaviors.keys(): raise "ERROR: beh already exists: '" + b.name + "'" else: self.behaviors[b.name] = b # keep a pointer to parent engine, from the beh: b.engine = self.engine b.brain = self.engine.brain b.robot = self.engine.robot # keep a pointer to parent state, from the beh: b.state = self b.setup() # init the behavior, just once if b.status: b.onActivate() def setcontrols(self, controls): self.controls = controls def run(self): for bkey in self.behaviors.keys(): b = self.behaviors[bkey] if b.status: b.rules = [] # clear rules b.update() # fires IF rules for r in b.rules: # r = truth, controller, amount, beh name, state name r.extend([b.name, b.state.name]) self.brain.desires.append( r ) # what is the controller effect for this state/behavior? self.brain.effectsTotal[b.state.name+":"+b.name+":"+r[1]] = b.effects[r[1]] self.update() def push(self, statename = None): if statename == None: statename = self.name if statename not in self.brain.states: raise AttributeError, "push: not a valid state name '%s'" % statename self.brain.stack.append( statename ) def pop(self): if len(self.brain.stack) > 0: returnState = self.brain.stack.pop() self.goto(returnState) else: raise AttributeError, "pop without a push in state '%s'" % self.name # wrappers here to talk to default robot: def move(self, *args): return self.robot.move(*args) def translate(self, *args): return self.robot.translate(*args) def rotate(self, *args): return self.robot.rotate(*args) def stop(self): return self.robot.stop() def startDevice(self, *args, **keywords): return self.robot.startDevice(*args, **keywords) def removeDevice(self, *args, **keywords): return self.robot.removeDevice(*args, **keywords) def motors(self, *args): return self.robot.motors(*args) def getDevice(self, *args): return self.robot.getDevice(*args) def hasA(self, *args): return self.robot.hasA(*args) def requires(self, *args): return self.robot.requires(*args) pyrobot/brain/behaviors/bare.py0000775000076500007650000000050210244170211016370 0ustar dblankdblank# A bare robot and brain from pyrobot.robot.simple import * class bareEngine: def __init__(self): self.status = 1 self.robot = SimpleRobot() def step(self): print "step!" def run(self, list = []): while self.status == 1: self.step() def init(): return bareEngine() pyrobot/brain/behaviors/CVS/0000775000076500007650000000000010273261045015550 5ustar dblankdblankpyrobot/brain/behaviors/CVS/Entries0000664000076500007650000000017710273261045017111 0ustar dblankdblank/__init__.py/1.29/Mon Jul 4 06:06:52 2005// /bare.py/1.2/Sun May 22 20:50:17 2005// /core.py/1.5/Sun May 22 20:50:17 2005// D pyrobot/brain/behaviors/CVS/Repository0000664000076500007650000000003010273261045017643 0ustar dblankdblankpyrobot/brain/behaviors pyrobot/brain/behaviors/CVS/Root0000664000076500007650000000003210273261045016411 0ustar dblankdblank:pserver:dblank@bubo:/cvs pyrobot/brain/behaviors/core.py0000775000076500007650000000221110244170211016406 0ustar dblankdblank# Some core/stock behaviors from pyrobot.brain.behaviors import * from math import sqrt class StopBehavior (Behavior): def init(self): self.Effects('translate', 1) self.Effects('rotate', 1) def update(self): self.IF(1, 'translate', 0) self.IF(1, 'rotate', 0) class ForwardBehavior (Behavior): def init(self): self.Effects('translate', 1) def update(self): self.IF(1, 'translate', 1) class BackwardBehavior (Behavior): def init(self): self.Effects('translate', 1) def update(self): self.IF(1, 'translate', -1) class TurnLeftBehavior (Behavior): def init(self): self.Effects('rotate', 1) def update(self): self.IF(1, 'rotate', 1) class TurnRightBehavior (Behavior): def init(self): self.Effects('rotate', 1) def update(self): self.IF(1, 'rotate', -1) class StraightBehavior (Behavior): def init(self): # called when created self.Effects('translate', .1) self.Effects('rotate', .1) def update(self): self.IF(1, 'translate', .2) self.IF(1, 'rotate', 0) pyrobot/brain/sequence1.py0000664000076500007650000000165110271002541015371 0ustar dblankdblank__author__ = "Douglas Blank " __version__ = "$Revision: 1.6 $" from pyrobot.brain.governor import GovernorSRN #from pyrobot.brain.conx import SRN mask = [0] * 4 + [0] * 5 + [1] * 4 net = GovernorSRN(delta = 0.3, epsilon = 2.1, historySize = 5, mask = mask) #net = SRN() net.addThreeLayers(4,5,4) net.setPatterns({"a" : [1,0,0,0], "b" : [0,1,0,0], "c" : [0,0,1,0], "d" : [0,0,0,1]}) net.setInputs( [["a", "b", "b", "b", "a", "d", "b"], ["a", "c", "c", "c", "a", "d", "c"]] ) #net.setInputs( [["a", "b", "a", "d", "b"], # ["a", "c", "a", "d", "c"]] ) net.predict("input", "output") net.setInitContext(0) net.setTolerance(.3) net.setStopPercent(0.75) net.setResetEpoch(2500) net.setResetLimit(1) net.setEpsilon(0.05) net.setMomentum(0) net.governing = 1 net.train() print net.ravq net.setLearning(0) net.setInteractive(1) net.sweep() pyrobot/brain/psom/0000775000076500007650000000000010273263701014112 5ustar dblankdblankpyrobot/brain/psom/makefile0000775000076500007650000000107010137575604015622 0ustar dblankdblank# simple makefile for maintaining a concurrent version of a built file # (i'm using it to manage a .so file) in multiple places, where you # want an up-to-date version in the directory with this makefile but # there is also a copy in the directory LIB, along with a makefile there # with a rule for building it, and a cleanall rule for, well, cleaining LIB=csom_src TARGET=_csom.so # ----------------------- $(TARGET): $(LIB)/$(TARGET) mv $(LIB)/$(TARGET) . $(LIB)/$(TARGET): (cd $(LIB) && make $(TARGET)) clean:: $(RM) $(TARGET) (cd $(LIB) && make clean) pyrobot/brain/psom/__init__.py0000775000076500007650000022355110271067717016245 0ustar dblankdblank# daniel sproul, june 24, 2002 # modified june 17, 2002 (yeelin tan) # psom is a python interface to csom.so, which is itself in turn a low-level # python interface to the som-pak c library. # this file implements the following classes: # psom # vector # dataset # point # activations from os import getenv import time, types from posixpath import exists import _csom as csom csom.set_globals() # neither worry about nor change this # doug, trust me, this sets global parameters that # are invariant from som to som class PsomError(AttributeError): """ Used to indicate that an attribute reference or assignment failed for a psom object. """ class VectorError(AttributeError): """ Used to indicate that an attribute reference or assignment failed for a vector object. """ class DatasetError(AttributeError): """ Used to indicate that an attribute reference or assignment failed for a dataset object. """ class PointError(AttributeError): """ Used to indicated that an attribute reference or assignment failed for a point object. """ class FileNotFound(IOError): """ Used to indicate that an attribute does not name an existing file. """ # Hacks added to replace the ptrset, ptrvalue, ptrcreate functions # from pointer.i which has been deprecated with SWIG 1.3. cpointer.i # is the new SWIG pointer library now. Need to find a better way to # determine C types. The reason Doug checks the last 3 or 5 elements # in myarr for type is because SWIG encodes pointers to primitive C # types such as int *, double ***, and char ** in a representation # that contains the actual value of the pointer and a type tag. For # example, the SWIG representation of the above pointers may look like # the following: _10081012_p_int, _1008e124_ppp_double, f8ac_pp_char. # These representations are of type str in python. def _ptrset(myarr, item, i): """ Given an array, an item, and an index, this function inserts the item into the array at position indexed by i. NOTE: Users should not ever use this function. """ print "FIXME:", myarr, item, i if myarr[-5:] == 'float': csom.floatarray_setitem(myarr, i, item) elif myarr[-5:] == 'short': csom.shortarray_setitem(myarr, i, item) elif myarr[-3:] == 'int': csom.intarray_setitem(myarr, i, item) elif myarr[-4:] == 'char': csom.charstararray_setitem(myarr, i, item) # array of character pointers else: raise TypeError, myarr def _ptrvalue(myarr, i): """ Given an array and an index, this function returns the value stored at index position i. NOTE: Users should not ever use this function. """ if myarr[-5:] == 'float': return csom.floatarray_getitem(myarr, i) elif myarr[-5:] == 'short': return csom.shortarray_getitem(myarr, i) elif myarr[-3:] == 'int': return csom.intarray_getitem(myarr, i) elif myarr[-4:] == 'char': return csom.charstararray_getitem(myarr, i) # array of character pointers else: raise TypeError, myarr def _ptrcreate(type, init, nitems): """ Given a type and the number of elements, this function creates an array of the specified type and number of elements. The init param is unused. NOTE: Users should not ever use this function. """ if type == "short": return csom.new_shortarray(nitems) elif type == "float": return csom.new_floatarray(nitems) elif type == "int": return csom.new_intarray(nitems) elif type == "char": return csom.new_charstararray(nitems) # array of character pointers else: raise TypeError, type csom.ptrcreate = _ptrcreate csom.ptrset = _ptrset csom.ptrvalue = _ptrvalue def list_to_arr(mylist,type): """ Function to convert a python list into a c array/pointer. NOTE: Users should not ever use this function. PARAMS mylist: python list type : type of list elements """ nitems = len(mylist) myarr = csom.ptrcreate(type,0,nitems) i = 0 for item in mylist: csom.ptrset(myarr,item,i) i += 1 return myarr def arr_to_list(myarr,nitems): """ Function to convert a c array/pointer into a python list. NOTE: Users should not ever use this function. PARAMS myarr: c array/pointer type : type of array elements """ """ WKV 2003-07-28 mylist = [] for i in range(nitems): mylist.append(csom.ptrvalue(myarr,i)) """ mylist = [0] * nitems for i in xrange(nitems): mylist[i] = csom.ptrvalue(myarr, i) return mylist ################################################################################# # class: psom # ################################################################################# class psom: def __init__(self, xdim='unset', ydim='unset', topol='hexa', neigh='gaussian', alpha_mode='linear', radius_mode='linear', rmin=0.0, rmax=1.0, dim='unset', data='unset', file='unset'): """ psom: __init__() ---------------- psom constructor In general, construct a psom, initialize it, then train/map to it, periodically doing something with the information it produces. PARAMS: xdim, ydim : x and y dimensions of the map topol : topology of the map ('hexa' or 'rect'. 'hexa' by default.) neigh : neighborhood type ('gaussian' or 'bubble'. 'gaussian' by default.) alpha_mode : function governing the decrease in learning rate ('linear' or 'inverse_t'. 'linear' by default.) radius_mode: function governing the decrease in the radius of learning effect. ('linear' or 'inverse_t'. 'linear' by default.) rmin,rmax : min and max values for random init of SOM model vectors (0.0 and 1.0 by default) dim : length of SOM model vectors data : dataset containing sample data vectors for SOM initialization file : .cod file containing model vectors to be loaded into SOM USAGE: 1. to read a som's model vectors from a .cod file (same file format as used in som_pak): >>> mysom = psom(file=filename) 2. to create a som with model vectors evenly distributed in the space of a pre-existing data set: >>> mysom = psom(xdim,ydim,data=mydataset) 3. to randomly initialize >>> mysom = psom(xdim,ydim,dim=d) - the values of the model vectors would be generated using the default rmin and rmax values which are 0.0 and 1.0 respectively . or >>> mysom = psom(xdim,ydim,dim=d,rmin=range_min,rmax=range_max) - the values of the model vectors would be generated using user specified rmin and rmax values. """ if(alpha_mode=='inverse_t'): alpha_mode = csom.INVERSE_T else: alpha_mode = csom.LINEAR if(radius_mode=='inverse_t'): radius_mode = csom.INVERSE_T else: radius_mode = csom.LINEAR if(file!='unset'): if exists(file): # som created from .cod file, so no need to specify dims codes = csom.open_entries(file) self.params = csom.construct_teach_params(codes, alpha_mode, radius_mode) else: raise FileNotFound, "File '%s' was not found" % file else: if(xdim=='unset' or ydim=='unset'): raise PsomError, "x and y dimensions for the map must be specified" if(topol=='rect'): topol=csom.TOPOL_RECT else: topol=csom.TOPOL_HEXA if(neigh=='bubble'): neigh=csom.NEIGH_BUBBLE else: neigh=csom.NEIGH_GAUSSIAN if(data=='unset'): # dataset unspecified; random init of model vectors if(dim=='unset'): raise PsomError, "Vector dimension must be specified for random init" data = dataset(dim=dim) # create a dataset with user specified vector dimension import random gen = random.Random() for count in range(10 * xdim * ydim): mylist = [] for i in range(dim): # random() returns a floating point number between 0 and 1 mylist.append(gen.random() * (rmax - rmin) + rmin) data.addvec(vector(mylist)) codes = csom.lininit_codes(data.data,topol,neigh,xdim,ydim) self.params = csom.construct_teach_params(codes,alpha_mode,radius_mode) self.xdim = csom.entries_xdim_get(codes) self.ydim = csom.entries_ydim_get(codes) self.dim = csom.entries_dimension_get(codes) # dim of model vectors if(csom.teach_params_topol_get(csom.teach_params_counters_teach_get \ (self.params)) == csom.TOPOL_RECT): self.topol = 'rect' else: self.topol = 'hexa' self.last = 'unset' # flag indicating SOM has not been trained; get_activations() will not work self.last_trained = 'unset' self.last_mapped = 'unset' self.logging = 0 self.log_mode = 'both' self.log_type = 'file' self.log_dir = getenv("PWD") + "/" self.log_fprefix = "psom" self.log_count = 0 self.log_file = 'unopened' self.log_padding = 4 self.log_format = "%C: [%i] maps to %p at time %d-%t\n" def __del__(self): """ psom: __del__() --------------- Added by WKV 2003-07-28 """ if self.params: if csom: codes = csom.teach_params_codes_get(csom.teach_params_counters_teach_get(self.params)) csom.close_entries(codes) csom.free_teach_params(self.params) self.params = None def init_training(self,alpha_0,radius_0,runlen,errorwindow=1): """ psom: init_training() --------------------- Sets the initial learning rate, radius of learning effect and the number of training patterns. runlen is the number of training samples you expect to use in the current training session. This number affects the rate at which alpha and radius decay. By default, the decay of these is linear, such that after running runlen samples, alpha is zero and radius is 1.0. It is kind of important to set unlen appropriately. Instead of using the default linear decay function, both radius and alpha can be set to decay with an inverse time function (i.e alpha_mode='inverse_t'...), which can sometimes be less drastic because then the learning rate never actually ends up at zero, even if you exceed the expected runlen number of training samples. you can also reinitialize a training session whenever you like (currently alpha_mode and radius_mode are set in stone when the psom is created, however) errorwindow is sort of an ad-hoc way of making the error value returned by get_error() time dependent on the last 'errorwindow' number of training samples. The default is 1, which should just return the error associated with the most recent training sample. PARAMS alpha_0 : initial learning rate. Range: 0 <= alpha_0 <= 1.0 radius_0 : initial radius of learning effect. Range: 0 <= radius_0 <= xdim or ydim (whichever is bigger) runlen : number of training patterns. Range: runlen >= 1 errorwindow: number of previous training samples to consider when calculating SOM error. Range: errorwindow >= 1 (1 by default). See get_error(). USAGE 1. To initialize a som with a learning rate of 0.02, radius of learning effect of 3, and number of training patterns of 1000: >>> mysom.init_training(0.02,3,1000) 2. Once the som is initialized, train it using one of the following: >>> mysom.train_from_dataset or >>> mysom.train(vector1) See train() and train_from_dataset(). 3. map() acts a lot like train() but will not actually adjust any map model vectors (equivalent to training with zero alpha). See map(). 4. To time a training session, see timing_start(). Note: error checking that was previously done in init_training_session() in som_devrobs.c has been moved here. """ # check alpha (learning rate) if(alpha_0 < 0.0 or alpha_0 > 1.0): raise PsomError, \ "Invalid learning rate: %s. Alpha must be between 0 and 1.0 \ (inclusive)" % alpha_0 # check radius of learning effect if(radius_0 < 1.0): raise PsomError, \ "Invalid radius: %s. Radius of learning effect must be at least 1" \ % radius_0 # make sure radius of learning effect is not greater than map dimension if(radius_0 > self.xdim and radius_0 > self.ydim): raise PsomError, \ "Invalid radius: %s. SOM dimension is only %sx%s" \ % (radius_0, self.xdim, self.ydim) # runlen must be greater than or equal to 1 if(runlen < 1): raise PsomError, \ "Invalid run length: %s. Run length must be at least 1" % runlen # error window must be greater than or equal to 1 if(errorwindow < 1): raise PsomError, \ "Invalid size of error window: %s. Window size must be at least 1" \ % errorwindow self.runlen = runlen csom.init_training_session(self.params,alpha_0,radius_0,runlen,errorwindow) def timing_start(self): """ psom: timing_start() -------------------- Start timing. USAGE: >>> mysom.timing_start() >>> mysom.train_from_dataset(mydataset) >>> mysom.timing_stop() >>> ttime = mysom.get_training_time() >>> print ttime """ csom.timing_start(self.params) def timing_stop(self): """ psom: timing_stop() ------------------- Stop timing. USAGE: >>> mysom.timing_start() >>> mysom.train_from_dataset(mydataset) >>> mysom.timing_stop() >>> ttime = mysom.get_training_time() >>> print ttime """ csom.timing_stop(self.params) def get_training_time(self): """ psom: timing_stop() ------------------- Returns training time if training session was timed. USAGE: >>> mysom.timing_start() >>> mysom.train_from_dataset(mydataset) >>> mysom.timing_stop() >>> ttime = mysom.get_training_time() >>> print ttime """ return csom.get_training_time(self.params) # psom: som logging utility def logging_set(self, mode='unset', type='unset', dir='unset', prefix='unset', padding='unset', format='unset'): """ psom: logging_set() ------------------- This function is used to initialize logging settings. PARAMS mode : specifies what kind of data vectors are logged -- those that are trained ('train'), those that are mapped ('map'), or 'both'. type : type can be 'dir' or 'file'; 'dir' will create a separate file for each log entry and 'file' will create a single file, putting each entry on a separate line. dir : directory for the target file(s) prefix: file prefix for the target file(s) format: specifies a format string: %i: input vector (entries separated by spaces) %m: model vector (entries separated by spaces) %p: coordinates of model vector as '(x,y)' %x: x coordinate of model vector %y: y coordinate of model vector %c: the current log count %C: a padded version of the current log count %d: the current date as 'MM/DD/YY' %D: date as 'Mon DD, YYYY' %e: date as 'MM/DD' %E: date as 'Mon DD' %t: the current 24-hour time as 'HH:MM:SS' %T: 12-hour time as 'HH:MM:SS:AM/PM' %u: 24-hour time as 'HH:MM' %U: 12-hour time as 'HH:MM:AM/PM' \n: newline \t: tab %%: a literal '%' sign """ if(mode != 'unset'): if(mode != 'map' and mode != 'train' and mode != 'both'): raise PsomError, \ "Unrecognized logging mode: %s" % mode self.log_mode = mode if(type != 'unset'): if(type != 'file' and type != 'dir'): raise PsomError, \ "Unrecognized logging type: %s" % type self.log_type = type if(dir != 'unset'): if(dir[-1] != "/"): dir += "/" if(dir[0:2] == "~/"): dir = getenv("HOME") + dir[1:] if(dir[0] != "/"): dir = getenv("PWD") + "/" + dir self.log_dir = dir if(prefix != 'unset'): self.log_fprefix = prefix if(padding != 'unset'): self.log_padding = padding if(format != 'unset'): self.log_format = format def logging_on(self): """ psom: logging_on() ------------------ Turn logging on. """ self.logging = 1 if(self.log_type == 'file'): fname = self.log_dir + self.log_fprefix + ".log" self.log_file = open(fname, "a") def logging_off(self): """ psom: logging_off() ------------------- Turn logging off. """ self.logging = 0 if(self.log_type == 'file'): self.log_file.close() def logging_reset(self): """ psom: logging_reset() --------------------- Reset the log count. """ self.log_count = 0 def logging_clear(self): """ psom: logging_clear() --------------------- Remove log files """ from os import system if(self.log_type == 'dir'): fname = self.log_dir + self.log_fprefix + "[0-9]*.log" system("rm " + fname) else: if(self.logging): self.log_file.close() fname = self.log_dir + self.log_fprefix + ".log" system("rm " + fname) if(self.logging): self.log_file = open(fname, "a") def log(self, model_vec): """ psom: log() ----------- PARAMS model_vec: model vector to be logged """ import re tag = "____%____%____" # this should be something no one will ever use padcnt = str(self.log_count) while(len(padcnt) < self.log_padding): padcnt = "0" + padcnt s = self.log_format s = re.sub("%%", tag, s) s = re.sub("%i", str(self.last), s) s = re.sub("%m", str(self.get_model_vector(self.last.point)), s) s = re.sub("%p", str(self.last.point), s) s = re.sub("%x", str(self.last.point.x), s) s = re.sub("%y", str(self.last.point.y), s) s = re.sub("%c", str(self.log_count), s) s = re.sub("%C", padcnt, s) s = re.sub("%t", time.strftime("%X"), s) s = re.sub("%T", time.strftime("%I:%M:%S:%p"), s) s = re.sub("%u", time.strftime("%H:%M"), s) s = re.sub("%U", time.strftime("%I:%M:%p"), s) s = re.sub("%d", time.strftime("%x"), s) s = re.sub("%D", time.strftime("%h %d, %Y"), s) s = re.sub("%e", time.strftime("%m/%d"), s) s = re.sub("%E", time.strftime("%h %d"), s) s = re.sub(tag, "%", s) if(self.log_type == 'dir'): fname = self.log_dir + self.log_fprefix + padcnt +".log" self.log_file = open(fname, "w") self.log_file.write(s) if(self.log_type == 'dir'): self.log_file.close() self.log_count += 1 def map(self,vector): """ psom: map() ----------- Given a psom vector, this function maps the vector to the SOM. The model vector of the winning node is returned. Model vectors of neighboring nodes remain unchanged, i.e no learning occurs. PARAMS vector: vector object to be mapped USAGE >>> mylist = [0,1,0,2] >>> mysom.map(psom.vector(mylist)) """ # make sure dimensions of mapping vector and SOM model vector match if(vector.dim != self.dim): raise PsomError, \ "Mismatched dimensions of mapping vector (size %s) \ and model vector (size %s)" % (vector.dim, self.dim) if self.last_mapped == 'unset': # first run coords = csom.map_one(self.params, vector.entry, point(-1,-1).asIntPtr(), 1) # 1=update counters else: coords = csom.map_one(self.params, vector.entry, self.last_mapped.asIntPtr(), 1) pt = point(csom.ptrvalue(coords,0), csom.ptrvalue(coords,1)) csom.delete_intarray(coords) # WKV 2003-07-28 self.last = vector self.last.point = pt self.last_mapped = pt model = self.get_model_vector(pt) if(self.logging and self.log_mode != 'train'): self.log(model) return model def train(self,vector): """ psom: train() ------------- Given a psom vector, this function maps the vector to the SOM. The model vector of the winning node is returned. Model vectors of neighboring nodes are changed so that their values are closer to that of the winnning node. PARAMS vector: vector to be mapped USAGE >>> mylist = [0,1,0,2] >>> mysom.train(psom.vector(mylist)) """ # make sure dimensions of training vector and SOM model vector match if(vector.dim != self.dim): raise PsomError, \ "Mismatched dimensions of training vector (size %s) \ and model vector (size %s)" % (vector.dim, self.dim) if self.last_trained == 'unset': # first run coords = csom.train_one(self.params, vector.entry, point(-1,-1).asIntPtr(), 1) # 1=update counters else: coords = csom.train_one(self.params, vector.entry, self.last_trained.asIntPtr(), 1) pt = point(csom.ptrvalue(coords,0), csom.ptrvalue(coords,1)) csom.delete_intarray(coords) self.last = vector self.last.point = pt self.last_trained = pt model = self.get_model_vector(pt) if(self.logging and self.log_mode != 'map'): self.log(model) return model def train_from_dataset(self,dataset,mode='cyclic'): """ psom: train_from_dataset() -------------------------- Trains SOM from vectors in a dataset. There are two training modes: cyclic and rand (for random). The dimension of the vectors in the dataset must match the dimension of the SOM model vectors. PARAMS dataset: dataset of training vectors mode : mode of training from dataset ('cyclic' or 'rand'. 'cyclic' by default, i.e. vectors are mapped in order, and training is repeated iteratively depending on the number of training patterns specifiied at initialization.) As for 'rand', vectors are chosen at random from the dataset. USAGE >>> mydataset = psom.dataset(file=filename) >>> mysom.train_from_dataset(mydataset) """ def train_from_dataset(self, dataset, mode='cyclic'): # make sure dimension of dataset vectors and SOM model vectors match if(dataset.dim != self.dim): raise PsomError, \ "Mismatched dimensions of training vector (size %s) \ and model vector (size %s)" % (dataset.dim, self.dim) if(mode == 'rand'): mode = csom.RAND else: mode = csom.CYCLIC entry = csom.train_fromdataset(self.params,dataset.data,mode) #self.last = vector(entry=entry,dim=self.dim) self.last = vector(entry=entry,dim=self.dim, label=csom.get_label_data_entry(entry, csom.data_entry_num_labs_get(entry))) coords = csom.map_one(self.params, self.last.entry, point(-1,-1).asIntPtr(), 0) # 0=don't update counters pt = point(csom.ptrvalue(coords,0), csom.ptrvalue(coords,1)) csom.delete_intarray(coords) # WKV 2003-07-28 self.last.point = pt self.last_trained = pt return self.get_model_vector(pt) def map_from_dataset(self,dataset): """ psom: map_from_dataset() ------------------------ Maps vectors from a dataset in order to the SOM. The dimension of the vectors in the dataset must match the dimension of the SOM model vectors. PARAMS dataset: dataset of vectors to be mapped USAGE >>> mydataset = psom.dataset(file=filename) >>> mysom.map_from_dataset(mydataset) """ # make sure dimension of dataset vectors and SOM model vectors match if(dataset.dim != self.dim): raise PsomError, \ "Mismatched dimensions of training vector (size %s) \ and model vector (size %s)" % (dataset.dim, self.dim) entry = csom.map_fromdataset(self.params, dataset.data) #self.last = vector(entry=entry,dim=self.dim) self.last = vector(entry=entry, dim=self.dim, label=csom.get_label_data_entry(entry, \ csom.data_entry_num_labs_get(entry))) coords = csom.map_one(self.params, self.last.entry, point(-1,-1).asIntPtr(), 0) # 0=don't update counters pt = point(csom.ptrvalue(coords,0),csom.ptrvalue(coords,1)) csom.delete_intarray(coords) # WKV 2003-07-28 self.last.point = pt self.last_mapped = pt return self.get_model_vector(pt) def get_model_vector(self, point): """ psom: get_model_vector() ------------------------ Returns model vector of SOM node at the specified point. PARAMS point: point object. see point constructor. USAGE >>> mysom.get_model_vector(psom.point(2, 3)) - this returns the model vector at node (2,3) of the SOM """ codes = csom.teach_params_codes_get(csom.teach_params_counters_teach_get(self.params)) entry = csom.get_model_vector(codes, point.asIntPtr()) label = csom.get_label_data_entry(entry, csom.data_entry_num_labs_get(entry)) dim = csom.entries_dimension_get(codes) return vector(entry=entry, dim=dim, point=point, label=label) def get_activations(self,mode='bubble',radius=1.0): """ psom: get_activations() ----------------------- Returns the som activation levels after the som has been mapped to. Calling this function prior to som mapping will result in error. For pretty print, call display_activations() on the return value of this function. After calling map(), train(), or train_from_dataset(), calling this function returns a list corresponding to the appropriate activation levels. This can be done based on simple 'bubble' or 'gaussian' neighborhoods, or (as suggested by Doug) by assigning activation weight according to error in mapping to each corresponding model vector ('error'). If the 'error' mode is used, the radius is taken as a tolerance parameter, and should vary between 0.0 to 1.0. 0.0 will look a lot more like bubble activation, while 1.0 will have pretty much every node activated to at least some degree. In summary, mode: bubble radius range: >= 1.0 mode: gaussian radius range: >= 1.0 mode: error radius range: between 0 and 1.0 (inclusive) PARAMS mode : 'bubble' (by default), 'gaussian', 'error' radius: radius of effect USAGE >>> myactiv = mysom.get_activations('bubble', 2.0) >>> myactiv = mysom.get_activations('bubble',2.0) >>> myactiv = mysom.get_activations('gaussian',3.0) >>> myactiv = mysom.get_activations('error') >>> myactiv = mysom.get_activations('error',0.25) Note: error values are always >= 0.0 """ if(self.last == 'unset'): raise PsomError, \ "Cannot get activation levels. SOM has not yet been mapped to." if(mode == 'gaussian' or mode == 'bubble'): if(mode == 'gaussian'): mode = csom.NEIGH_GAUSSIAN else: mode = csom.NEIGH_BUBBLE float_levels = csom.get_activation_levels(self.params, self.last.point.asIntPtr(), radius, mode) elif(mode == 'error'): float_levels = csom.get_levels_by_error(self.params, self.last.entry,radius) else: raise PsomError, "mode %s has not yet implemented" % mode levels = arr_to_list(float_levels,self.xdim*self.ydim) csom.delete_floatarray(float_levels) # WKV 2003-07-28 return levels def get_xy(self, mylist, x, y): """ psom: get_xy() -------------- This function allows x,y addressing of a list representation of a matrix. PARAMS mylist: list representation of a matrix (e.g. activation levels of som nodes returned by get_activations()) x : x coordinate of matrix element y : y coordinate of matrix element USAGE To obtain the activation level of som node at coordinate (2,3): >>> mysom.get_xy(get_activations(),2,3) """ return mylist[x + y * self.xdim] def save_to_file(self, file): """ psom: save_to_file() -------------------- Given a filename, this function saves the SOM model vectors to a specified file. Use this function to save the model vectors after a training session. PARAMS file: name of file. recommended extension: .cod USAGE >>> mysom.save_to_file(filename.cod) """ codes = csom.teach_params_codes_get(csom.teach_params_counters_teach_get(self.params)) csom.write_entries(codes,file) def display(self): """ psom: display() --------------- Displays all model vectors in the given som. USAGE >>> mysom.display() """ codes = csom.teach_params_codes_get(csom.teach_params_counters_teach_get(self.params)) csom.print_dataset(codes) def display_activations(self,levels='unset'): """ psom: display_activations() --------------------------- Displays som activations in pretty print. If the levels parameter is unspecified, then, by default, this function displays the activation levels returned by calling get_activations() with gaussian mode and a radius of 1.0. Otherwise, the user can specify the activation levels to be displayed by calling get_activations() with desired mode and radius. PARAMS levels: list of activation levels. USAGE 1. To use default display (gaussian mode and radius = 1.0): >>> mysom.display_activations() 2. To display the activation levels returned by get_activations() with gaussian mode and radius of 3: >>> mysom.display_activations(get_activations(mode='gaussian', radius=3)) """ if(levels == 'unset'): levels = self.get_activations('gaussian') for j in range(0,self.ydim): if(j % 2 == 1 and self.topol=='hexa'): print " ", for i in range(0,self.xdim): index = i + j * self.xdim print "%.2f" % (levels[index]), print "" def get_reg_counter(self, pt, mode): """ psom: get_reg_counter() ----------------------- Given a point and a mode ('train' or 'map'), this function returns the corresponding (regular) counter associated with that point. PARAMS pt : point object mode: 'train' or 'map' USAGE: 1. To get training (regular) counter associated with point (0,2) >>> mysom.get_counter(point(0,2), 'train') 2. To get mapping (regular) counter associated with point (5,5) >>> mysom.get_counter(point(5,5), 'map') """ if not isinstance(pt, point): raise TypeError, "Expecting a point object, given %s intead" % type(pt) if mode == 'train': return csom.get_reg_tcounter(self.params, pt.asIntPtr()) elif mode == 'map': return csom.get_reg_mcounter(self.params, pt.asIntPtr()) else: raise PsomError, "Unknown mode: %s" % mode def get_consec_counter(self, pt, mode): """ psom: get_consec_counter() -------------------------- Given a point and a mode ('train' or 'map'), this function returns the corresponding consecutive counter associated with that point. PARAMS pt : point object mode: 'train' or 'map' USAGE: 1. To get consecutive training counter associated with point (0,2) >>> mysom.get_consec_counter(point(0,2), 'train') 2. To get consecutive mapping counter associated with point (5,5) >>> mysom.get_consec_counter(point(5,5), 'map') """ if not isinstance(pt, point): raise TypeError, "Expecting a point object, given %s intead" % type(pt) if mode == 'train': return csom.get_consec_tcounter(self.params, pt.asIntPtr()) elif mode == 'map': return csom.get_consec_mcounter(self.params, pt.asIntPtr()) else: raise PsomError, "Unknown mode: %s" % mode def get_maxconsec_counter(self, pt, mode): """ psom: get_consec_counter() -------------------------- Given a point and a mode ('train' or 'map'), this function returns the corresponding maximum consecutive counter associated with that point. PARAMS pt : point object mode: 'train' or 'map' USAGE: 1. To get maximum consecutive training counter associated with point (0,2) >>> mysom.get_max_consec_counter(point(0,2), 'train') 2. To get maximum consecutive mapping counter associated with point (5,5) >>> mysom.get_max_consec_counter(point(5,5), 'map') """ if not isinstance(pt, point): raise TypeError, "Expecting a point object, given %s intead" % type(pt) if mode == 'train': return csom.get_maxconsec_tcounter(self.params, pt.asIntPtr()) elif mode == 'map': return csom.get_maxconsec_mcounter(self.params, pt.asIntPtr()) else: raise PsomError, "Unknown mode: %s" % mode def get_highest_counter(self, counter_type='unset', mode='unset'): """ Given a mode ('train' or 'map') and a counter type ('regular' -- default, 'consec', 'max_consec'), this function looks for the specified counter with the highest value, and returns a reference to the corresponding point object and the counter in a two-element list. Return format: [pt, counter value] USAGE: 1. To get point and highest (regular) train counter: >>> mysom.get_highest_counter(mode='train') 2. To get point and highest consecutive train counter: >>> mysom.get_highest_counter('consec', 'map') 3. To get point and highest max consecutive map counter: >>> mysom.get_highest_counter('max_consec', 'train') """ """ if mode == 'train': dict = self.train_counter_dict elif mode == 'map': dict = self.map_counter_dict else: raise PsomError, "Unknown mode: %s" % mode if counter_type == 'consec': ct_type = 1 elif counter_type == 'max_consec': ct_type = 2 else: ct_type = 0 max_val = 0 max_ls = [] for i in range(self.xdim): for j in range(self.ydim): pt = point(i,j) curr_val = dict[pt.__str__()][ct_type] if curr_val >= max_val: if curr_val > max_val: max_val = curr_val max_ls = [] max_ls.append([pt, curr_val]) return max_ls """ """ max = 0 for i in range(self.xdim): for j in range(self.ydim): pt = point(i,j) counter = self._get_highest_counter_helper(counter_type, pt, mode) if counter > max: max = counter return max def _get_highest_counter_helper(self, counter_type, pt, mode): if counter_type == 'consec': self.get_consec_counter(pt, mode) elif counter_type == 'maxconsec': self.get_maxconsec_counter(pt, mode) else: self.get_reg_counter(pt, mode) """ ################################################################################# # class: vector # ################################################################################# class vector: """ Used both for data and model vectors. get_elts() can be used to access the actual values of the entries in the vector. The point data member should remain 'unset' for data vectors, but for model vectors represents its coordinates in the SOM """ def __init__(self, elts='unset', weight=1, mask='NULL', entry='unset', dim='unset', point='unset', label=[]): """ vector: __init__() ------------------ vector constructor PARAMS elts : list of elements to be converted into a c float array weight: training weight of the vector (1 by default). greater weights have greater training impact on the map. weight must be >= 0. mask : list of 1's and 0's. length of mask list should be the same as that of elts list. a 1 indicates that the corresponding value in elts list would be ignored in computing the winning model vector. if all values in the elts list are to be considered in the computation of the winning model vector, then the mask should be NULL or set to [0,0,...,0]. entry : reserved for internal use. after initialization, this param stores the SWIG representation of the array pointer. dim : reserved for internal use. after initialization, this param stores the size of the c array. point : this param should remain 'unset' for data vectors, but for model vectors this represents its coordinates in the som. label : list of integers or strings to be associated with a vector. e.g. ['X'], [1, 45], ['ab', 'C3'], etc. Note: Typically, all a user need worry about is the elts parameter. USAGE 1. The vector constuctor is typically used in the following way: >>> mysom.train(vector([7,2.5,0.2])) 2. To use masking: >>> mysom.train(vector([7,2.5,0.2],mask=[1,0,0]) -- the first element of the vector (i.e. 7) will be ignored when computing the winning model vector. 3. To map a vector with a label: >>> mysom.map(vector([7,2.5,0.2],label=['X',2,'ab','C4'])) """ if(elts != 'unset'): # make sure weight of training vector >= 0. # Note: This check used to be performed in som_devrobs.c if(weight < 0.0): raise VectorError, \ "Invalid weight for training vector: %s. \ Weight must be >= 0" % weight points = list_to_arr(elts, "float") dim = len(elts) # mask c_mask = 'NULL' # Warning: c_mask must be initialized to NULL, otherwise call to # make_data_entry_weighted_masked() below will fail if mask is NULL. if(mask != 'NULL'): # make sure len(mask) matches vector dimension if (len(mask) != dim): raise VectorError, \ "Mismatched dimensions of vector (len %s) and mask (len %s)" \ % (dim,len(mask)) # make sure mask is a binary list for item in mask: if(item != 0 and item != 1): raise VectorError, \ "Invalid mask: %s. Mask must be a binary list" % mask c_mask = list_to_arr(mask, "short") # convert mask list into a c array of shorts # label #c_str_label = 'NULL' # Warning: c_str_label must be initialized to NULL first. str_label = [] if(len(label) != 0): # convert every elt in label (python list) to a string, # storing it in str_label, a new python list of strings. #str_label = [] for item in label: str_label.append(str(item)) # Null-terminate the list of strings. this is needed to determine # the array size in make_data_entry_weighted_masked() in som_devrobs.c #str_label.append(str(0)) # Convert list of strings into a c array of character pointers. # Note: each elt in c_str_label is a char pointer. So, c_str_label is the # equivalent of C's char **. #c_str_label = list_to_arr(str_label, "char") # create a data entry struct (see som_devrobs.c for more info) entry = csom.make_data_entry_weighted_masked(points, weight, c_mask, dim, str_label) # data members of vector object (separated from code for clarity) self.dim = dim self.entry = entry self.point = point self.mask = mask def get_elts(self): """ vector: get_elts() ------------------ Returns the original list representation of the elements in the vector. """ points = csom.data_entry_points_get(self.entry) return arr_to_list(points, self.dim) def __getitem__(self, key): """ vector: __getitem__() --------------------- Returns the vector element at index position given by key. """ mylist = self.get_elts() if isinstance(key, types.SliceType): print key.start print key.stop #WKV 2003-07-28 if key.stop > len(self): stop = len(self) else: stop = key.stop return mylist[key.start:stop] #return mylist[key.start:key.stop] # WKV 2003-07-28 else: return mylist[key] def __len__(self): """ vector: __len__() ----------------- Returns the length of the vector. """ return len(self.get_elts()) def get_weight(self): """ vector: get_weight() -------------------- Returns the training weight of the vector. See vector constructor for more info. """ return csom.data_entry_weight_get(self.entry) def get_mask(self): """ vector: get_mask() ------------------ Returns the mask of the vector. Mask is returned as a python list. """ return self.mask # DEPRECATED """ def set_label(self,label=[]): vector: set_label() ------------------- Associates a vector with the given label. Removes all labels previously associated with the vector. #c_str_label = 'NULL' # Warning: c_str_label must be initialized to NULL first. #if(label == 'NULL'): if(len(label) == 0): raise VectorError, "No label provided" # convert every elt in label (python list) to a string, # storing it in str_label, a new python list of strings. str_label = [] for item in label: str_label.append(str(item)) # Null-terminate the list of strings. this is needed to determine # the array size in make_data_entry_weighted_masked() in som_devrobs.c str_label.append(str(0)) # Convert list of strings into a c array of character pointers. # Note: each elt in c_str_label is a char pointer. So, c_str_label is the # equivalent of C's char **. #c_str_label = list_to_arr(str_label, "char") #csom.set_label_data_entry(self.entry, c_str_label) print str_label csom.set_label_data_entry(self.entry, str_label) self.label = str_label """ #def add_label(self,label='NULL'): def add_label(self,label=[]): """ vector: add_label() ------------------- Associates a vector with the given label. All labels previously associated with the vector remain unchanged. """ #c_str_label = 'NULL' # Warning: c_str_label must be initialized to NULL first. #if(label == 'NULL'): if(len(label) == 0): raise VectorError, "No label provided" # convert every elt in label (python list) to a string, # storing it in str_label, a new python list of strings. str_label = [] for item in label: str_label.append(str(item)) # Null-terminate the list of strings. this is needed to determine # the array size in make_data_entry_weighted_masked() in som_devrobs.c #str_label.append(str(0)) # Convert list of strings into a c array of character pointers. # Note: each elt in c_str_label is a char pointer. So, c_str_label is the # equivalent of C's char **. #c_str_label = list_to_arr(str_label, "char") #csom.add_label_data_entry(self.entry, c_str_label) csom.add_label_data_entry(self.entry, str_label) def clear_label(self): """ vector: clear_label() --------------------- Removes all labels associated with the vector. For some reason, this is giving me a segfault. """ csom.clear_labels_data_entry(self.entry) def get_label(self): """ vector: get_label() ------------------- Returns the label list associated with the vector or 'NULL' if no labels are associated with this vector. """ return csom.get_label_data_entry(self.entry, self.get_numlabels()) def get_label_asString(self): """ vector: get_label_asString() ---------------------------- Returns the list of labels associated with the vector in a string representation. Used in vis.py for labeling of SOM nodes in gui. """ label_ls = self.get_label() label_str = "" if len(label_ls) != 0: for item in label_ls: label_str += item return label_str def get_numlabels(self): """ Returns the number of labels associated with the vector. """ return csom.data_entry_num_labs_get(self.entry) def __str__(self): """ vector: __str__() ----------------- Returns a string representation of all elements in the vector. This function is typically called by print(). Only the elements in the vector are returned, i.e. weight, mask, and label are not. """ mylist = self.get_elts() """ # WKV 2003-07-28 s = "" for elt in mylist: s += "%.3f" % (elt) + " " s = s[:-1] return s """ mylist = ["%.3f" % x for x in mylist] return " ".join(mylist) def display(self): """ vector: display() ----------------- Displays the vector as a list of values. Datamembers (e.g. weight, mask, and label) associated with the vector are displayed as well. Format: [] weight: mask:[] label:[